package ltq.telphoneAgeAverage;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;
import ltq.util.*;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.HashMap;
import java.util.Map;

public class TAABolt extends BaseRichBolt {
    private Map<Bean1, Bean2> ageAverageList;
    private int item,itemB;
    Connection conn=null;
    PreparedStatement ps = null;
    ResultSet rs = null;
    @Override
    public void prepare(Map<String, Object> map, TopologyContext topologyContext, OutputCollector outputCollector) {
        ageAverageList = new HashMap<>();
        item=0;
        itemB=0;
        conn= DatabaseUtil.getConnection();
    }

    @Override
    public void execute(Tuple tuple) {
        String brand=tuple.getStringByField("brand");
        if (brand.equals("null")){
            System.out.println("-----null------");
            for (Bean1 key : ageAverageList.keySet()) {
                refreshData(key,ageAverageList.get(key));
            }
            System.out.println(item);
            return;
        }
        int age=Integer.parseInt(tuple.getStringByField("age"));
        String sex=tuple.getStringByField("sex");
        itemB=Integer.parseInt(tuple.getStringByField("flag"));

        Bean1 bean1A =new Bean1(brand,sex);
        Bean2 bean2A=new Bean2(1,age);
        boolean flag2=false;
        for (Bean1 key : ageAverageList.keySet()) {
            if(key.getBrand().equals(brand)&&key.getSex().equals(sex)){
                Bean2 bean2=ageAverageList.get(key);
                bean2.setAge(bean2.getAge()+age);
                bean2.setNum(bean2.getNum()+1);
                flag2=true;
                break;
            }
        }
        if (!flag2){
            ageAverageList.put(bean1A,bean2A);
        }
        item+=1;
        System.out.println("item1:"+item);

        if (item%itemB==0){
            for (Bean1 key : ageAverageList.keySet()) {
                refreshData(key,ageAverageList.get(key));
            }
        }
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer outputFieldsDeclarer) {

    }

    public void refreshData(Bean1 bean1, Bean2 bean2){
        try {
            ps= conn.prepareStatement("select * from TAAresult where rBrand=? and rSex=?");
            ps.setObject(1,bean1.getBrand());
            ps.setObject(2,bean1.getSex());
            rs= ps.executeQuery();
            if (rs.next()){
                //数据已存在，update
                ps= conn.prepareStatement("update TAAresult set rAverageAge=? where rBrand=? and rSex=?");
                ps.setObject(1,(double)bean2.getAge()/(double)bean2.getNum());
                ps.setObject(2,bean1.getBrand());
                ps.setObject(3,bean1.getSex());
                ps.executeUpdate();
            }
            else {
                //数据不存在，insert
                ps= conn.prepareStatement("INSERT into TAAresult(rBrand,rSex,rAverageAge) values (?,?,?)");
//                System.out.println("-----num------"+num);
                ps.setObject(1,bean1.getBrand());
                ps.setObject(2,bean1.getSex());
                ps.setObject(3,(double)bean2.getAge()/(double)bean2.getNum());
                ps.executeUpdate();
            }
        } catch (SQLException e) {
            throw new RuntimeException(e);
        }
    }
}
